
package kvsnode;

import java.io.FileInputStream;
import java.io.IOException;
import java.rmi.AccessException;
import java.rmi.AlreadyBoundException;
import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.rmi.registry.LocateRegistry;
import java.rmi.registry.Registry;
import java.rmi.server.UnicastRemoteObject;
import java.util.Collection;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import util.Log;
import util.LogWriter;
import util.Validator;

/**
 * This class is an actual KVS node, and simulator of the virtual KVS nodes. It
 * has a responsibility to create, maintain, and fix KVS nodes(actual and
 * virtual) in a server. It can be also considered as an actual KVS node in a
 * server by implementing KVSNode.
 * 
 * The node port number for the constructor is used for the port number of an
 * actual node, and it will increment the original port number to use them for
 * the virtual nodes.
 * 
 * @author cy
 * @version 5/3/2013
 * 
 */
public class ActualKVSNodeImpl extends AbstractKVSNode
{
  /**
   * To save memory.
   */
  public static final String YES = "y";

  /**
   * To save memory.
   */
  public static final String NO = "n";

  /**
   * Wait time for the actual node at the initializing phase.
   */
  public static final int WAIT_TIME = 20000;

  /**
   * Wait time to make sure all the nodes in the ring are up. We use 1.5 as a
   * multiplier to make sure all the nodes are up because it should be bigger
   * than the WAIT_TIME, and We are not sure that how much it will be taking to
   * create the virtual nodes for the actual nodes, but we think 1.5 is pretty
   * reasonable number to guarantee that all the nodes are up in this simple
   * local network application.
   */
  public static final int SUCCESSOR_FINDING_WAIT_TIME = (int) (WAIT_TIME * 1.5);

  /**
   * The internal(fake) RMI Remote registry.
   */
  private static Registry static_registry;

  /**
   * It has information of the KVS nodes this server deals with (including the
   * virtual nodes). This field is a static to avoid garbage collection.
   */
  private final Map<Integer, KVSNode> my_node_map = new HashMap<Integer, KVSNode>();

  /**
   * List of the successor actual nodes in increasing order.
   */
  private final List<Integer> my_successors = new LinkedList<Integer>();

  /**
   * A constructor for actual node.
   * 
   * - related instruction in the instruction document -
   * 
   * The actual nodes in the system are registered with a single RMI registry;
   * this is analogous to hosts on the Internet being registered with the Domain
   * Name System (DNS).Each group will use its own RMI registry. For the sake of
   * non-interference during development and testing, consider ports 80XX (where
   * XX is a team number) reserved for RMI registries; for example, team 03
   * would use port 8003 for their registry on whatever machine they choose to
   * run it on. Despite this reservation, your system should be implemented such
   * that it can use an RMI registry on any host and port. Nodes are named in
   * the RMI registry such that they can be looked up by number (e.g., by
   * appending the number to a fixed string), because nodes in the ring will
   * need to find their successors and neighbors in the registry.
   * 
   * @param the_properties the configuration object for the node.
   */
  public ActualKVSNodeImpl(final Properties the_properties)
  {
    super(the_properties);
    locateRMIRegistry();
  }

  /**
   * Locate the RMI registry.
   */
  private void locateRMIRegistry()
  {
    // If the node is to create a new KVS system, the return value is "y"; if
    // not, n.
    final String build_kvs = getProperties().getProperty("buildKVS");
    // if this node is the first actual nod in the ring, return "y"; if not, n.
    final String is_first_node = getProperties().getProperty("isfirstnode");
    final String hostname = getProperties().getProperty("rmihost");
    final String rmi_port = getProperties().getProperty("rmiport");
    try
    {

      // Check if a valid port number.
      if (Validator.isPortNumber(rmi_port))
      {
        final int rmi_port_int = Integer.valueOf(rmi_port);
        if (YES.equals(build_kvs) && YES.equals(is_first_node))
        {
          // Create registry on the specified port.
          final Registry registry = LocateRegistry.createRegistry(rmi_port_int);
          static_registry = new RemoteRegistry();
          registry.bind(REMOTE_REGISTRY, UnicastRemoteObject.exportObject(static_registry, 0));
          Log.out("RMI registry listening on port " + rmi_port);
        }
        else
        {
          final Registry registry = LocateRegistry.getRegistry(hostname, rmi_port_int);
          // the casting is only possible for interface. It is impossible to
          // cast to RemoteRegistry
          static_registry = (Registry) registry.lookup(REMOTE_REGISTRY);
          Log.out("RMI registry listening on host: " + hostname + " and port: " + rmi_port);
        }
        Log.out("Remote registry bound with name '" + REMOTE_REGISTRY + "'");
      }
      else
      {
        Log.err("port must be between 1025 and 65535 inclusive");
        System.exit(1);
      }
    }
    catch (final AccessException e)
    {
      Log.err("issue accessing registry: " + e);
      System.exit(1);
    }
    catch (final AlreadyBoundException e)
    {
      Log.err("already bound???: " + e);
      System.exit(1);
    }
    catch (final RemoteException e)
    {
      Log.err("a problem occurred: " + e);
      System.exit(1);
    }
    catch (final ArrayIndexOutOfBoundsException | NumberFormatException e)
    {
      Log.err("Invalid start parameters" + e);
      Log.out(e.toString());
    }
    catch (final NotBoundException e)
    {
      Log.err("Could not find RMI server.");
      System.exit(1);
    }
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void runNode()
  {
    final String build_kvs = getProperties().getProperty("buildKVS");
    if (YES.equals(build_kvs))
    {
      startUp();
      simulateVirtualNode();
      findSuccessors();
    }
    else
    {
      requestRegistration();
    }
  }

  /**
   * Locate an actual node in the ring at the initialized stage. It going
   * through the ring clockwise to find empty spot from 0, and each time adding
   * 1 to the node id to find next location. It is not a perfect solution to
   * find an empty spot, but we assume that we start the ring with few actual
   * nodes.
   */
  private void startUp()
  {
    final String node_id = getProperties().getProperty("nodeid");
    
    // Check for valid node id.
    if (Validator.isInteger(node_id) && Integer.valueOf(node_id) >= 0)
    {
      int i = Integer.valueOf(node_id);
      boolean not_found = true;
      while (i < getTotalNodeNumber() && not_found)
      {
        try
        {
          static_registry.bind("" + i + "", UnicastRemoteObject.exportObject(this, 0));
          
          // Set to be an actual node and set id.
          setActualNodeId(i);
          setNodeId(i);
          
          // Add node to map.
          my_node_map.put(i, this);
          not_found = false;
          
          // Find "m" variable.
          final int m = Integer.valueOf(getProperties().getProperty("m"));
          LogWriter.logStartup(i, m);
          // Find successor actual nodes, then write them down on the log file.
          Log.out("bound to " + i);
        }
        catch (final RemoteException e)
        {
          Log.err(e.toString());
          System.exit(1);
        }
        catch (final AlreadyBoundException e)
        {
          Log.out("node " + i + " is already bound, checking next node.");
          i++;
        }
      }
      if (not_found)
      {
        Log.out("There is no spot for you. Sorry.");
        System.exit(1);
      }
    }
  }

  /**
   * This is to simulate the virtual nodes at the initial stage.
   * 
   * - related instruction in the instruction document -
   * 
   * The system is started with 2^m nodes (each of which is responsible for
   * storing the values for 1/2^m keys), numbered 0 . . . 2^m - 1. Both m and
   * its own number are known to each node when it starts up. Some of the nodes
   * in the system are virtual and others are actual. For example, a 16 node
   * system might be started with only 4 actual nodes, numbered 2, 5, 9, and 14;
   * each of these actual nodes is responsible for simulating other virtual
   * nodes such that there is a full set of 2^m nodes. With no redundancy, each
   * actual node might be responsible for simulating the nodes numbered higher
   * than it and lower than the next actual node (modulo 2^m ). Thus, actual
   * node 2 would also simulate virtual nodes 3-4, actual node 5 would also
   * simulate virtual nodes 6-8, actual node 9 would also simulate virtual nodes
   * 10- 13, and actual node 14 would also simulate virtual nodes 15, 0, and 1.
   * Later, when you need to implement fault tolerance, you will want each node
   * to be simulated by more than one actual node (so that if one actual node
   * disappears, no data is lost).
   */
  private void simulateVirtualNode()
  {
    Log.out("Wait " + WAIT_TIME + " millisec to let other actual node can come in.");
    try
    {
      // wait certain amount of time until it generates virtual nodes.
      Thread.sleep(WAIT_TIME);
    }
    catch (final InterruptedException e1)
    {
      e1.printStackTrace();
    }
    Log.out("Generating virtual nodes for actual node " + getNodeId());
    int i = getNextNodeInRing(getNodeId());
    boolean unoccupied_node = true;
    while (i != getNodeId() && i < getTotalNodeNumber() && unoccupied_node)
    {
      try
      {
        // Create new virtual node.
        final VirtualKVSNodeImpl temp_node =
            new VirtualKVSNodeImpl(getProperties(), i, getNodeId());
        
        // Bind node, add to map.
        static_registry.bind("" + i + "", UnicastRemoteObject.exportObject(temp_node, 0));
        my_node_map.put(i, temp_node);
        temp_node.runNode();
        i = getNextNodeInRing(i);
      }
      catch (final RemoteException e)
      {
        Log.err(e.toString());
        unoccupied_node = false;
        System.exit(1);
      }
      catch (final AlreadyBoundException e)
      {
        Log.out("Found bound node, node: " + i);
        unoccupied_node = false;
      }
    }
  }

  /**
   * Find the successor actual nodes for this actual node in clock order, then
   * put them into my_successors List.
   */
  private void findSuccessors()
  {
    Log.out("Im waiting " + SUCCESSOR_FINDING_WAIT_TIME + " millisec to find succssors");
    try
    {
      // wait certain amount of time until it generates virtual nodes.
      Thread.sleep(SUCCESSOR_FINDING_WAIT_TIME);
    }
    catch (final InterruptedException e1)
    {
      e1.printStackTrace();
    }
    int i = getNextNodeInRing(getNodeId());
    try
    {
      // Look for all actual nodes in the registry.
      while (i != getNodeId())
      {
        final KVSNode node = (KVSNode) static_registry.lookup("" + i + "");

        // When an actual node is found, add to List.
        if (node.getActualNodeForThisNode().intValue() == i)
        {
          my_successors.add(i);
        }
        i = getNextNodeInRing(i);
      }
    }
    catch (final AccessException e)
    {
      Log.err("Successor finding: issue accessing registry: " + e);
      System.exit(1);
    }
    catch (final RemoteException e)
    {
      Log.err("Successor finding: a problem occurred: " + e);
      System.exit(1);
    }
    catch (final NotBoundException e)
    {
      Log.err("Successor finding: Could not find RMI server.");
      System.exit(1);
    }
    String successors = "";
    for (Integer node : my_successors)
    {
      successors += Integer.toString(node) + " ";
    }
    if (LogWriter.logSuccessor(getNodeId(), successors))
    {
      Log.out("Wrote successors to file.");
    }
    else
    {
      Log.err("Error writing successors to file.");
    }

  }

  /**
   * This method is to request registration to the one of the actual Node stub
   * in the RMI registry for the node id, and how many KVS nodes this server
   * should generate for the entire KVS system.
   * 
   * The return value is a list of integer having the node IDs it should
   * generate. The first element is for an actual node, from the second, it is
   * for virtual nodes. If it returns an empty list, the DNS is full of actual
   * nodes, so there is no spot for the request. For that reason, this
   * KVSNodeController cannot create any KVS node.
   */
  private void requestRegistration()
  {
    // To Do.
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public boolean importData(final Map<Integer, Collection<Object>> the_node_data_map)
  {
    // To Do.
    return false;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public Collection<Object> exportData(final int the_node_id)
  {
    // To Do.
    return null;
  }

  /**
   * Main method to start an actual node. It gets only one command line
   * parameter, the location of a config.properties file.
   * 
   * @param the_strings var args
   */
  public static void main(final String... the_strings)
  {
    final Properties properties = new Properties();
    try
    {
      // Load the properties file.
      properties.load(new FileInputStream(the_strings[0]));
      final KVSNode actual_node = new ActualKVSNodeImpl(properties);
      actual_node.runNode();
    }
    catch (final IOException e)
    {
      System.err.println("You should have a legal properties file as a "
                         + "commendline parameter for this application");
    }
  }

}
